package ex.datastream.windows;

import ex.datastream.functions.sourceFunction.SourceFuncByEvent;
import ex.vo.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;

/**
 * ref:https://blog.csdn.net/wangpei1949/article/details/102996170
 */
public class TimeEvictorTest {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.addSource(new SourceFuncByEvent())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<Event>() {
                                            @Override
                                            public long extractTimestamp(Event element, long recordTimestamp) {
                                                return element.timestamp;
                                            }
                                        }
                                )
                )
                .keyBy(event -> true)
                //1分钟的滑动窗口，只保留最近10内的数据
                .window(TumblingEventTimeWindows.of(Time.minutes(1)))
                .evictor(TimeEvictor.of(Time.seconds(10)))
                .process(new ProcessWindowFunction<Event, Object, Boolean, TimeWindow>() {
                    @Override
                    public void process(Boolean aBoolean, Context context, Iterable<Event> iterable, Collector<Object> collector) throws Exception {
                        SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        System.out.printf("start:%s, end:%s\n",dateformat.format(context.window().getStart()), dateformat.format(context.window().getEnd()));
                        collector.collect("窗口中有" + iterable.spliterator().getExactSizeIfKnown() + "条元素");
                        iterable.forEach(a-> System.out.println(a.toString()+","));
                        System.out.println("\n*******************************************************");
                    }
                })
                .print();

        env.execute();
    }
//start:2024-02-29 17:53:00, end:2024-02-29 17:54:00
//窗口中有9条元素
//Event{key='33', timestamp=1709200431484', date=2024-02-29 17:53:51},
//Event{key='49', timestamp=1709200432489', date=2024-02-29 17:53:52},
//Event{key='15', timestamp=1709200433506', date=2024-02-29 17:53:53},
//Event{key='14', timestamp=1709200434512', date=2024-02-29 17:53:54},
//Event{key='68', timestamp=1709200435522', date=2024-02-29 17:53:55},
//Event{key='62', timestamp=1709200436531', date=2024-02-29 17:53:56},
//Event{key='87', timestamp=1709200437540', date=2024-02-29 17:53:57},
//Event{key='47', timestamp=1709200438553', date=2024-02-29 17:53:58},
//Event{key='59', timestamp=1709200439565', date=2024-02-29 17:53:59},
//
//*******************************************************
//start:2024-02-29 17:54:00, end:2024-02-29 17:55:00
//窗口中有10条元素
//Event{key='44', timestamp=1709200490144', date=2024-02-29 17:54:50},
//Event{key='65', timestamp=1709200491159', date=2024-02-29 17:54:51},
//Event{key='46', timestamp=1709200492176', date=2024-02-29 17:54:52},
//Event{key='62', timestamp=1709200493191', date=2024-02-29 17:54:53},
//Event{key='82', timestamp=1709200494207', date=2024-02-29 17:54:54},
//Event{key='6', timestamp=1709200495224', date=2024-02-29 17:54:55},
//Event{key='79', timestamp=1709200496235', date=2024-02-29 17:54:56},
//Event{key='46', timestamp=1709200497239', date=2024-02-29 17:54:57},
//Event{key='29', timestamp=1709200498251', date=2024-02-29 17:54:58},
//Event{key='8', timestamp=1709200499272', date=2024-02-29 17:54:59},
//
//*******************************************************

}

